package ru.nsu.fit.iconveyor.engine.smart;

import ru.nsu.fit.iconveyor.conveyor.Conveyor;
import ru.nsu.fit.iconveyor.engine.exception.ChannelOverflowException;
import ru.nsu.fit.iconveyor.conveyor.node.*;
import ru.nsu.fit.iconveyor.conveyor.channel.Channel;
import ru.nsu.fit.iconveyor.conveyor.channel.ChannelDataProvider;
import ru.nsu.fit.iconveyor.engine.AbstractEngine;
import ru.nsu.fit.iconveyor.engine.channel.ChannelFactory;
import ru.nsu.fit.iconveyor.engine.channel.ConveyorChannelDataProvider;
import ru.nsu.fit.iconveyor.engine.exception.EndlessLoopException;

import java.util.*;

/**
 * Thread pool engine.
 *
 * @author Timur Zolotuhin (tzolotuhin@gmail.com)
 */
public class SmartEngine extends AbstractEngine {
    // Main data
    private State state = State.NOT_STARTED;
    private Conveyor conveyor;
    private int maxNodeToInputsLaunchesRatio;
    private int inputNodesLaunchesNumber = 0;
    private Map<String, Integer> launchesOfNodes = new HashMap<String, Integer>();

    private static final int DEFAULT_MAX_ITERATIONS_NUMBER = 1000;

    private Executor.Observer executorsObserver = new Executor.Observer() {
        @Override
        public void onError(Exception cause) {
            notifyEngineStateObserversOnError(cause, false);
        }
    };

    private Executor [] executors;
    private  Thread [] threads;
    private int threadNumber = 1;

    private Map<String, Queue<Cell4Result>> notOrderResults; // nodeId -> queue of cell of results
    private Map<String, Queue<Map<String, Object>>> notOrderInputs; // nodeId -> queue of cell of inputs

    private Queue<String> alreadyRunNodes; // contains node ids, which isParallel = false and it's already run

    // Mutex
    private final Object generalMutex = new Object();

    public SmartEngine(int threadNumber) {
        if (threadNumber <= 0) {
            threadNumber = 1;
        }

        this.threadNumber = threadNumber;
    }

    private ChannelFactory channelFactory = new ChannelFactory() {
        @Override
        public Channel newChannel(String id) {
            return new SimpleChannel(id);
        }
    };

    @Override
    public void setConveyor(Conveyor conveyor) {
        if (conveyor == null) {
            throw new IllegalArgumentException("conveyor can't be null");
        }
        if (this.conveyor != null) {
            throw new IllegalStateException("conveyor was already set");
        }
        this.conveyor = conveyor;
    }

    @Override
    public void start() {
        if (state != State.NOT_STARTED) {
            throw new IllegalStateException("illegal state of engine");
        }
        if (conveyor == null) {
            throw new IllegalStateException("conveyor was not set");
        }
        if (threadNumber <= 0) {
            throw new IllegalArgumentException("thread number can't be less or equal zero");
        }

        notifyEngineStateObserversOnStart();
        state = State.RUNNING;

        // create data provider and fiction channels
        ChannelDataProvider dataProvider = new ConveyorChannelDataProvider(conveyor, channelFactory);
        notOrderResults = new HashMap<String, Queue<Cell4Result>>();
        notOrderInputs = new HashMap<String, Queue<Map<String, Object>>>();
        alreadyRunNodes = new ArrayDeque<String>();

        for (Node node : conveyor.getNodes()) {
            notOrderResults.put(node.getId(), new ArrayDeque<Cell4Result>());
            notOrderInputs.put(node.getId(), new ArrayDeque<Map<String, Object>>());
        }

        maxNodeToInputsLaunchesRatio = calculateMaxNodeToInputsLaunchesRatio(conveyor);


        // create executors
        executors = new Executor[threadNumber];
        threads = new Thread[threadNumber];
        for (int i = 0; i < threadNumber; i++) {
            executors[i] = new Executor(i, generalMutex);
            executors[i].addObserver(executorsObserver);
            threads[i] = new Thread(executors[i]);
            threads[i].start();
        }

        try {
            // main loop
            boolean  error = false;
            while(!error) {

                boolean  hasExecutedNodes = false;
                Executor availableExecutor = getAvailableExecutor();

                // wait if available thread = null
                if (availableExecutor == null) {
                    try {
                        synchronized (generalMutex) {
                            generalMutex.wait();
                        }
                    } catch (InterruptedException ex) {
                        break;
                    }
                    continue;
                }

                Collection<Node> nodes = conveyor.getNodes();
                for (Node node : nodes) {
                    // check node, isParallel
                    if ((!node.isParallelExecute() || node.isSingleExecute()) && alreadyRunNodes.contains(node.getId()))
                        continue;

                    // collect input parameters
                    Collection<NodeConnection<?>> nodeConnections = node.getNodeConnections();
                    if (nodeConnections != null) {
                        Map<String, Object> bundles = new HashMap<String, Object>();

                        for (NodeConnection<?> nodeConnection : nodeConnections) {
                            if (nodeConnection.getType() == NodeConnection.Type.INPUT) {
                                if (nodeConnection.getChannelId() != null && !dataProvider.isChannelEmpty(nodeConnection.getChannelId())) {
                                    Object obj = dataProvider.peekFromChannel(nodeConnection.getChannelId(), nodeConnection.getDataClass());
                                    bundles.put(nodeConnection.getName(), obj);
                                }
                            }
                        }

                        // try to run new task
                        Map<String, Object> needBundles = node.preExecute(bundles, conveyor.getProperties());
                        if (needBundles != null) {
                            for (String nodeNameArg : needBundles.keySet()) {
                                if (needBundles.get(nodeNameArg) != null) {
                                    String channelId = node.getInputNodeConnectionByName(nodeNameArg).getChannelId();
                                    dataProvider.pollFromChannel(channelId, needBundles.get(nodeNameArg).getClass());
                                }
                            }

                            if(node instanceof InputNode) {
                                ++inputNodesLaunchesNumber;
                            } else {
                                Integer launchesOfNode = launchesOfNodes.get(node.getId());
                                if(launchesOfNode == null) {
                                    launchesOfNode = 0;
                                }

                                if(launchesOfNode > inputNodesLaunchesNumber * maxNodeToInputsLaunchesRatio) {
                                    throw new EndlessLoopException(String.format("found %d launches of node %s", launchesOfNode, node.getId()));
                                }

                                launchesOfNodes.put(node.getId(), ++launchesOfNode);
                            }

                            Cell4Result cell4Result = new Cell4Result(node.getId());
                            Executor.Task task = new Executor.Task();
                            task.setNode(node);
                            task.setInput(needBundles);
                            task.setCell4Result(cell4Result);
                            task.setProp(conveyor.getProperties());
                            try {
                                availableExecutor.addTask(task);
                                //System.err.println("Executor = " + availableExecutor.getNumber());
                                if (!node.isParallelExecute() || node.isSingleExecute())
                                    alreadyRunNodes.add(node.getId());
                            } catch (IllegalStateException e) {
                                throw new ChannelOverflowException("channel is overflown", e);
                            }
                            notOrderResults.get(node.getId()).add(cell4Result);
                            notOrderInputs.get(node.getId()).add(needBundles);
                            hasExecutedNodes = true;

                            availableExecutor = getAvailableExecutor();
                            if (availableExecutor == null) {
                                break;
                            }
                        }
                    }
                }

                // try synchronise data provider
                boolean hasNotOrderResults = false;
                for (Node node : nodes) {
                    Queue<Cell4Result> cell4Results = notOrderResults.get(node.getId());
                    Queue<Map<String, Object>> inputs = notOrderInputs.get(node.getId());
                    while (!cell4Results.isEmpty()) {
                        Cell4Result cell4Result = cell4Results.peek();
                        if (cell4Result.isComplete()) {
                            Map<String, Object> result = cell4Result.getResult();
                            if (result != null) {
                                for (String nodeNameArg : result.keySet()) {
                                    String channelId = node.getOutputNodeConnectionByName(nodeNameArg).getChannelId();
                                    if (channelId != null) {
                                        dataProvider.addToChannel(channelId, result.get(nodeNameArg));
                                    } else {
                                        error = true;
                                        notifyEngineStateObserversOnError(new Exception("Channel = null for OUTPUT ARGUMENT =  " + nodeNameArg + " for node = " + node.getId()), true);
                                    }
                                }
                            }

                            // call notifyConveyorExecutionObserversOnNodeExecuted
                            Collection<String> affectedInputs = new ArrayDeque<String>();
                            Collection<String> affectedOutputs = new ArrayDeque<String>();
                            if (inputs.peek() != null) {
                                for (String nodeNameArg : inputs.peek().keySet()) {
                                    if (inputs.peek().get(nodeNameArg) != null)
                                        affectedInputs.add(nodeNameArg + " = " + inputs.peek().get(nodeNameArg).toString());
                                }

                            }
                            if (cell4Result.getResult() != null) {
                                for (String nodeNameArg : cell4Result.getResult().keySet()) {
                                    if (cell4Result.getResult().get(nodeNameArg) != null) {
                                        affectedOutputs.add(nodeNameArg + " = " + cell4Result.getResult().get(nodeNameArg));
                                    }
                                }
                            }
                            notifyConveyorExecutionObserversOnNodeExecuted(node.getId(), affectedInputs, affectedOutputs);

                            // remove from not order queue
                            hasNotOrderResults = true;
                            cell4Results.poll();
                            inputs.poll();
                            if (!node.isParallelExecute() && !node.isSingleExecute())
                                alreadyRunNodes.remove(node.getId());
                        } else {
                            break;
                        }
                    }
                    hasNotOrderResults |= (cell4Results.size() > 0);
                }

                // check stop conveyor
                if (!hasExecutedNodes && isAllAvailableExecutor() && !hasNotOrderResults) {
                    notifyEngineStateObserversOnFinish();
                    state = State.FINISHED;
                    interruptThreads();
                    break;
                }
            }
        } catch(Exception e) {
            fail(e);
            e.printStackTrace();
        }
    }

    private static int calculateMaxNodeToInputsLaunchesRatio(Conveyor conveyor) {
        int maxIterationsNumber = DEFAULT_MAX_ITERATIONS_NUMBER;
        Object maxItarationsNumberString = conveyor.getProperties().get(Conveyor.PROP_MAX_ITERATIONS_NUMBER);
        if(maxItarationsNumberString != null)
            try {
                maxIterationsNumber = Integer.valueOf(maxItarationsNumberString.toString());
            } catch (NumberFormatException e) {
            }
        int maxInputsToOutputsRatio = 1;
        for(Node node : conveyor.getNodes()) {
            int ratio = node.getOutputToInputRatio();
            if(ratio < 1) {
                ratio = 1;
            }
            maxInputsToOutputsRatio *= ratio;
        }
        return maxInputsToOutputsRatio * maxIterationsNumber;
    }

    @Override
    public void stop() {
        if(state != State.RUNNING) {
            throw new IllegalStateException("illegal state of engine");
        }

        interruptThreads();

        notifyEngineStateObserversOnStop();
        state = State.STOPPED;
    }

    private void interruptThreads() {
        for (Thread thread : threads) {
            thread.interrupt();
        }
    }

    private void fail(Exception cause) {
        notifyEngineStateObserversOnError(cause, true);
        interruptThreads();
        state = State.FAILED;
    }

    @Override
    public State getState() {
        return state;
    }

    //==============================================================================
//-----------------PRIVATE METHODS----------------------------------------------
    private Executor getAvailableExecutor() {
        for (int i = 0; i < threadNumber; i++) {
            if (executors[i].isAvailable())
                return executors[i];
        }
        return null;
    }

    private boolean isAllAvailableExecutor() {
        for (int i = 0; i < threadNumber; i++) {
            if (!executors[i].isAvailable())
                return false;
        }
        return true;
    }

    private Stack<String> getOutputChannelIds(Node node) {
        Collection<NodeConnection<?>> nodeConnections = node.getNodeConnections();
        Stack<String> outputChannelIds = new Stack<String>();
        for (NodeConnection<?> nodeConnection : nodeConnections) {
            if (nodeConnection.getType() == NodeConnection.Type.OUTPUT)
                outputChannelIds.add(nodeConnection.getChannelId());
        }
        return  outputChannelIds;
    }

}
